package common

import (
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/kinwyb/appTools/log"
	"github.com/nsqio/go-nsq"
	"github.com/sirupsen/logrus"
	"io/ioutil"
	"math/rand"
	"net/http"
	"strings"
	"time"
)

// NSQMessage NSQ消息处理
type NSQMessage struct {
	nsqlookupdAddr    string                   //nsqlookupd地址
	config            *nsq.Config              //配置
	prefix            string                   //topic名称前缀
	log               *logrus.Entry            //日志
	clients           []*NSQMessageClient      //消费客户端
	producerBroadcast map[string]*nsq.Producer //消费端
	producers         []string                 //producers地址
	topicBroadcast    map[string]string        //话题后段地址
}

// NewNSQMessage 新建一个NSQMessage对象
func NewNSQMessage(nsqlookupdAddr string, prefix string, conf *nsq.Config) (*NSQMessage, error) {
	nsqlookupdAddr = strings.TrimSpace(nsqlookupdAddr)
	if nsqlookupdAddr == "" {
		return nil, errors.New("nsqlookupd地址空")
	}
	ret := &NSQMessage{
		nsqlookupdAddr: nsqlookupdAddr,
		prefix:         prefix,
		config:         conf,
		log:            log.WithField("tag", "nsq"),
	}
	err := ret.init()
	return ret, err
}

func (n *NSQMessage) init() error {
	if n.config == nil {
		n.config = nsq.NewConfig()
		n.config.LookupdPollInterval = 10 * time.Second
	}
	for _, v := range n.producerBroadcast {
		if v != nil {
			v.Stop()
		}
	}
	rand.Seed(time.Now().UnixNano())
	n.producerBroadcast = map[string]*nsq.Producer{}
	n.topicBroadcast = map[string]string{}
	n.getNSQDAddr()
	if n.clients != nil && len(n.clients) > 0 {
		oldClients := n.clients
		n.clients = make([]*NSQMessageClient, 0)
		for _, client := range oldClients {
			client.stop()
			n.NewClient(client.topic, client.channel, client.handler)
		}
	}
	return nil
}

func (n *NSQMessage) Reload() {
	n.init()
}

// Put 推送一个消息
func (n *NSQMessage) Put(topic string, msg []byte, delay ...time.Duration) error {
	producer, err := n.getProducer(topic)
	if err != nil {
		return err
	}
	for i := 0; i < 5; i++ { //如果发生错误再次推送，重复5次结束
		if delay == nil || len(delay) < 1 {
			err = producer.Publish(n.prefix+topic, msg)
		} else { //定时发送
			err = producer.DeferredPublish(n.prefix+topic, delay[0], msg)
		}
		if err != nil {
			n.log.Errorf("NSQ消息发送失败:%s", err.Error())
			continue
		}
		err = nil
		break
	}
	return err
}

// Output 输出到日志
func (n *NSQMessage) Output(calldepth int, s string) error {
	ss := strings.SplitN(s, " ", 2)
	if len(ss) < 2 {
		ss = append(ss, "")
	}
	ss[1] = strings.TrimSpace(ss[1])
	ms := strings.SplitN(ss[1], " ", 2)
	if len(ms) > 1 {
		ss[1] = strings.TrimSpace(ms[1])
	}
	switch ss[0] {
	case "INF":
		n.log.Info(ss[1])
	case "WRN":
		n.log.Warning(ss[1])
	case "ERR":
		n.log.Error(ss[1])
	default:
		//Log.Debug(ss[1])
	}
	return nil
}

// SetPrefix 设置前缀
func (n *NSQMessage) SetPrefix(prefix string) {
	n.prefix = prefix
}

// SetAddress 设置地址
func (n *NSQMessage) SetAddress(address string) {
	if n.nsqlookupdAddr != address {
		n.nsqlookupdAddr = address
	}
	n.Reload()
}

func (n *NSQMessage) getNSQDAddr() error {
	url := "http://" + n.nsqlookupdAddr + "/nodes"
	n.log.Debugf("查询nsqd地址：%s", url)
	resp, err := http.Get(url)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return err
	}
	nsqNode := &nsqNode{}
	json.Unmarshal(data, nsqNode)
	for _, v := range nsqNode.Producers {
		// 生成nsqd地址
		addr := fmt.Sprintf("%s:%d", v.BroadcastAddress, v.TcpPort)
		n.log.Debugf("查询到nsqd地址：%s", addr)
		for _, topic := range v.Topics { //nsqd地址区分话题
			n.topicBroadcast[topic] = addr
		}
		n.producers = append(n.producers, addr)
	}
	return nil
}

func (n *NSQMessage) getProducer(topic string) (*nsq.Producer, error) {
	var ret *nsq.Producer
	if broadcast, ok := n.topicBroadcast[topic]; ok {
		if producer, ok := n.producerBroadcast[broadcast]; ok {
			ret = producer
		}
	}
	if ret == nil {
		n.getNSQDAddr()
		index := rand.Int31n(int32(len(n.producers)))
		broadcast := n.producers[index]
		if localBroadcast, ok := n.topicBroadcast[topic]; ok {
			if producer, ok := n.producerBroadcast[localBroadcast]; ok {
				return producer, nil
			}
			broadcast = localBroadcast
		}
		producer, err := nsq.NewProducer(broadcast, n.config)
		if err != nil {
			n.log.WithError(err).WithField("info", broadcast).Error("NSQ新地址连接异常")
			return nil, err
		}
		producer.SetLogger(n, nsq.LogLevelDebug)
		n.producerBroadcast[broadcast] = producer
		n.topicBroadcast[topic] = broadcast
		n.log.Debugf("查询到 TOPIC: %s      NSQD:%s", topic, broadcast)
		ret = producer
	}
	return ret, nil
}

type nsqProducers struct {
	RemoteAddress    string   `json:"remote_address,omitempty"`
	Hostname         string   `json:"hostname,omitempty"`
	BroadcastAddress string   `json:"broadcast_address"`
	TcpPort          int64    `json:"tcp_port"`
	HttpPort         int64    `json:"http_port"`
	Topics           []string `json:"topics"`
}

type nsqNode struct {
	Producers []*nsqProducers `json:"producers,omitempty"`
}

// SetLog 设置日志
func (n *NSQMessage) SetLog(lg *logrus.Entry) {
	n.log = lg
	for _, c := range n.clients {
		c.setLogger(lg)
	}
}

// NewClient 创建客户端
func (n *NSQMessage) NewClient(topic string, channel string, handler nsq.Handler) error {
	n.config.MaxInFlight = 200
	client, err := newNSQMessageClient(
		n.nsqlookupdAddr,
		n.prefix+topic,
		channel,
		n.config)
	if err != nil {
		return err
	}
	client.setLogger(n.log)
	if handler != nil {
		client.setHandler(handler)
	}
	//连接NSQ
	if client.nsqlookupdAddress != "" {
		err = client.client.ConnectToNSQLookupd(client.nsqlookupdAddress)
		if err != nil {
			return err
		}
	} else {
		return errors.New("nsqlookupd地址空")
	}
	n.clients = append(n.clients, client)
	return nil
}

// NSQMessageClient NSQ消息处理客户端
type NSQMessageClient struct {
	client            *nsq.Consumer //消费
	nsqlookupdAddress string        //NSQLOOKUPD地址
	topic             string        //话题
	channel           string        //通道
	handler           nsq.Handler   //处理方式
	log               *logrus.Entry //日志
}

func newNSQMessageClient(nsqlookupdAddress string, topic string, channel string, conf *nsq.Config) (*NSQMessageClient, error) {
	ret := &NSQMessageClient{
		nsqlookupdAddress: nsqlookupdAddress,
		topic:             topic,
		channel:           channel,
		log:               log.WithField("tag", "nsq"),
	}
	var err error
	ret.client, err = nsq.NewConsumer(ret.topic, channel, conf)
	if err != nil {
		return nil, fmt.Errorf("%s-%s消息队列监听失败:%s", ret.topic, channel, err.Error())
	}
	ret.client.SetLogger(ret, nsq.LogLevelDebug)
	return ret, nil
}

func (n *NSQMessageClient) setHandler(handler nsq.Handler) {
	if n.client != nil {
		n.handler = handler
		n.client.AddHandler(handler)
	}
}

func (n *NSQMessageClient) setLogger(lg *logrus.Entry) {
	n.log = lg
}

func (n *NSQMessageClient) stop() {
	if n.client != nil {
		n.client.Stop()
		n.client = nil
	}
}

// Output 输出到日志
func (n *NSQMessageClient) Output(calldepth int, s string) error {
	ss := strings.SplitN(s, " ", 2)
	if len(ss) < 2 {
		ss = append(ss, "")
	}
	ss[1] = strings.TrimSpace(ss[1])
	ms := strings.SplitN(ss[1], " ", 2)
	if len(ms) > 1 {
		ss[1] = strings.TrimSpace(ms[1])
	}
	switch ss[0] {
	case "INF":
		n.log.Info(ss[1])
	case "WRN":
		n.log.Warning(ss[1])
	case "ERR":
		n.log.Error(ss[1])
	default:
		//Log.Debug(ss[1])
	}
	return nil
}
